/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.activity.impl;

import org.sabayframework.activity.ActivityContext;
import org.sabayframework.channel.Channel;
import org.sabayframework.channel.Lock;

/**
 * Stateful activity that reads all available input, stores it for processing and
 * after processing writes all result objects to output when available. 
 * @author Alex Antonau
 * */
public abstract class BufferedActivity<I, O> extends StatefulActivity<I, O> {

	private Object[] bufferedInput;
	private Object[] bufferedOutput;
	private int outputIndex;
	
	protected BufferedActivity() {
		super();
	}

	@Override
	public final void execute(ActivityContext<I, O> ctx) throws Exception {
		State state = getState(ctx);
		if(state == State.READY){
			final Channel<I> input = ctx.getInput();
			int inputSize = input.size();
			if(inputSize == 0) return;
			Lock<I> readLock = input.getReadLock(inputSize);
			while(inputSize > 1 && readLock==null){
				inputSize/=2;
				readLock = input.getReadLock(inputSize);
			}
			if(readLock!=null){
				try{
					bufferedInput = readLock.toArray();
				} finally{
					readLock.unlock();
				}
				setState(ctx, State.PROCESS);
			}else return;
			state = getState(ctx);
		}
		if(state == State.PROCESS){
			try{
				process(ctx);
			} finally{
				setState(ctx, State.OUTPUT);
			}
			state = getState(ctx);
		}
		if(state==State.OUTPUT){
			int outputSize = bufferedOutput == null?0:bufferedOutput.length-outputIndex; 
			if(outputSize > 0){
				final Channel<O> output = ctx.getOutput();
				final int available = output.freeSize();
				if(available == 0) return;
				if(outputSize > available) outputSize = available; 
				Lock<O> writeLock = output.getWriteLock(outputSize);
				while(outputSize > 1 && writeLock==null){
					outputSize/=2;
					writeLock = output.getWriteLock(outputSize);
				}
				if(writeLock!=null){
					try{
						output(ctx, writeLock);
					} finally{
						writeLock.unlock();
					}
				}
			}else setState(ctx, State.READY);
		}
	}

	@Override
	public final int getOutputSize(ActivityContext<I, O> ctx) {
		return 1;
	}

	@Override
	public final int getInputSize(ActivityContext<I, O> ctx) {
		return 1;
	}

	@Override
	public final void process(ActivityContext<I, O> ctx) throws Exception {
		bufferedOutput = process(ctx, bufferedInput);
		bufferedInput = null;
		outputIndex = 0;
	}
	/**
	 * Implementor should write body of this method.
	 * @param ctx Activity context
	 * @param bufferedInput array of input objects. Size of the array can vary but always > 0.
	 * @return array of output objects or null if no output is required.   
	 * */
	public abstract Object[] process(ActivityContext<I, O> ctx, Object[] bufferedInput) throws Exception;

	@Override
	public final void input(ActivityContext<I, O> ctx, Lock<I> inputLock){}

	@SuppressWarnings("unchecked")
	@Override
	public final void output(ActivityContext<I, O> ctx, Lock<O> outputLock)	throws Exception {
		int size = outputLock.size();
		for(int i=0;i<size;++i){
			outputLock.set(i, (O) bufferedOutput[outputIndex]);
			bufferedOutput[outputIndex++] = null; 
		}
		if(bufferedOutput.length == outputIndex){
			bufferedOutput = null;
			outputIndex = 0;
			setState(ctx, State.READY);
		}
	}

}
